package com.study.classification

import org.apache.spark.ml.classification.LogisticRegression
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.max

/**
 * 分类-逻辑回归(二分类)
 *
 * @author stephen
 * @date 2019-08-27 14:46
 */
object BinomialLogisticRegressionDemo {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()

    spark.sparkContext.setLogLevel("warn")

    import spark.implicits._

    val path = this.getClass.getClassLoader.getResource("data/mllib/sample_libsvm_data.txt").getPath
    val data = spark.read.format("libsvm").load(path)
    // 切分训练集和测试集，使用固定的种子来开始生成随机数。在给定相同的种子下，生成的随机数序列总是相同的。
    val Array(trainingData, testData) = data.randomSplit(Array(0.8, 0.2), seed = 1234L)

    // 二项逻辑回归
    val lr = new LogisticRegression()
      .setMaxIter(10) // 设置最大迭代次数(默认100)，具体迭代过程可能会在不足最大迭代次数时停止
      .setRegParam(0.3) // 设置正则化项系数(默认0.0)，正则化项主要用于防止过拟合现象
      .setElasticNetParam(0.8) // 正则化范式比(默认0.0)，正则化一般有两种范式：L1(Lasso)和L2(Ridge)。L1一般用于特征的稀疏化，L2一般用于防止过拟合。这里的参数即设置L1范式的占比，默认0.0即只使用L2范式
      .setTol(1E-7) // 设置容错(默认1E-6)，每次迭代会计算一个误差值，误差值会随着迭代次数的增加逐渐减小，如果误差值小于设置的容错值，则停止迭代优化
      .setLabelCol("label") // 设置标签列(默认读取“label”列)
      .setFeaturesCol("features") // 设置特征列(默认读取“features”列)

    // 训练模型
    val lrModel = lr.fit(trainingData)
    // 打印系数和截距
    println(s"CoefficientMatrix is: ${lrModel.coefficientMatrix}")
    println(s"InterceptVector is: ${lrModel.intercept}")

    // 使用测试集验证
    val predictions = lrModel.transform(testData)
    predictions.show()


    // 多项逻辑回归
    val mlr = new LogisticRegression()
      .setMaxIter(10)
      .setRegParam(0.3)
      .setElasticNetParam(0.8)
      .setFamily("multinomial")
    // 训练模型
    val mlrModel = mlr.fit(trainingData)
    // 打印多项逻辑回归的系数和截距
    println(s"Multinomial coefficients: ${mlrModel.coefficientMatrix}")
    println(s"Multinomial intercepts: ${mlrModel.interceptVector}")


    // 评估方式一
    // 获得回归模型训练的Summary
    val trainSummary = lrModel.binarySummary

    // 获得每次迭代的损失，一般会逐渐减小
    val objectiveHistory = trainSummary.objectiveHistory
    println("objectiveHistory:")
    objectiveHistory.foreach(loss => println(loss))

    // 得到ROC
    val roc = trainSummary.roc
    // 显示ROC数据表，可以用这个数据自己画ROC曲线
    roc.show()
    // AUC
    println(s"areaUnderROC: ${trainSummary.areaUnderROC}")

    // 不同的阈值，计算不同的F1，然后通过最大的F1找出并重设模型的最佳阈值。
    val fMeasure = trainSummary.fMeasureByThreshold
    // 获得最大的F1值
    val maxFMeasure = fMeasure.select(max("F-Measure")).head().getDouble(0)
    // 找出最大F1值对应的阈值（最佳阈值）
    val bestThreshold = fMeasure.where($"F-Measure" === maxFMeasure)
      .select("threshold").head().getDouble(0)
    // 并将模型的Threshold设置为选择出来的最佳分类阈值
    lrModel.setThreshold(bestThreshold)
    println(s"bestThreshold: ${bestThreshold}")


    // 评估方式二：二分类评估器
    val evaluator = new BinaryClassificationEvaluator()
      .setLabelCol("label")
      .setRawPredictionCol("prediction")
    // areaUnderROC|areaUnderPR
    val areaUnderROC = evaluator.setMetricName("areaUnderROC").evaluate(predictions);
    val areaUnderPR = evaluator.setMetricName("areaUnderPR").evaluate(predictions);

    println("============= BinaryClassificationEvaluator ===============")
    println(s"areaUnderROC is $areaUnderROC")
    println(s"areaUnderPR is $areaUnderPR")

    spark.stop()
  }
}
